package com.fwmagic.flink.projectcase.streamjoin;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.concurrent.TimeUnit;

public class StreamDataLeftSource extends RichParallelSourceFunction<Tuple3<String, String, Long>> {

    private volatile boolean running = true;

    @Override
    public void run(SourceContext<Tuple3<String, String, Long>> ctx) throws Exception {
        //准备数据
        Tuple3[] elements = new Tuple3[]{
                Tuple3.of("a", "1", 10050000L), //[50000-60000)
                Tuple3.of("a", "2", 10054000L), //[50000-60000)
                Tuple3.of("a", "3", 10079900L), //[[70000-80000))
                Tuple3.of("a", "4", 10115000L), //[[110000-120000)) //115000 - 5000 = 110000(触发[[100000-110000))窗口，后面两条延迟的数据丢失),115000 - 5002 = 109998[[[100000-110000))窗口没有被触发，后面两条数据在程序停止后会被触发计算]）
                Tuple3.of("b", "5", 10100000L), //[[100000-110000))
                Tuple3.of("b", "6", 10108000L), //[[100000-110000))
        };

        int count = 0;
        while (running && count < elements.length) {
            //将数据发送出去
            ctx.collect(Tuple3.of((String) elements[count].f0, (String) elements[count].f1, (Long) elements[count].f2));
            count++;
            TimeUnit.SECONDS.sleep(1);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
